package com.atguigu.gmall.realtime.test;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author Felix
 * @date 2023/11/11
 * 该案例演示了FlinkSQL基本连接
 *                              左表                      右表
 *      内连接             onCreateAndWrite        onCreateAndWrite
 *      左外连接           onReadAndWrite(续期)     onCreateAndWrite
 *      右外连接           onCreateAndWrite        onReadAndWrite(续期)
 *      全外连接           onReadAndWrite(续期)     onReadAndWrite(续期)
 */
public class Flink06_FlinkSql {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(2);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从指定的网络端口读取员工数据，并转换为动态表
        SingleOutputStreamOperator<Emp> empDS = env
            .socketTextStream("hadoop102", 8888)
            .map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Emp(Integer.valueOf(fieldArr[0]), fieldArr[1], Integer.valueOf(fieldArr[2]), Long.valueOf(fieldArr[3]));
                }
            );
       tableEnv.createTemporaryView("emp",empDS);

        //TODO 4.从指定的网络端口读取部门数据，并转换为动态表
        SingleOutputStreamOperator<Dept> deptDS = env
            .socketTextStream("hadoop102", 9999)
            .map(
                lineStr -> {
                    String[] fieldArr = lineStr.split(",");
                    return new Dept(Integer.valueOf(fieldArr[0]), fieldArr[1], Long.valueOf(fieldArr[2]));
                }
            );

        tableEnv.createTemporaryView("dept",deptDS);

        //TODO 5.内连接
        //注意：如果使用普通的内外连接，在底层会为参与连接的两张表各自维护一个状态
        //默认情况下，状态永不失效
        //在生产环境中，一定要设置状态的失效时间  tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10))
        // tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e join dept d on e.deptno = d.deptno").print();

        //TODO 6.左外连接
        //如果左表数据先到，右表数据后到的时候会产生3条数据
        //当左表数据到来的时候，生成一条数据    [左表 null] 标记为+I
        //当右表数据到来的时候，先生成一条数据   [左表 null] 标记为-D
        //再生成一条 [左表 右表] 标记为+I
        // 这样的动态表转换为的流称之为回撤流
        // tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno").print();

        //TODO 7.右外连接
        // tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e right join dept d on e.deptno = d.deptno").print();

        //TODO 8.全外连接
        // tableEnv.executeSql("select e.empno,e.ename,d.deptno,d.dname from emp e full join dept d on e.deptno = d.deptno").print();

        //TODO 9.将左外连接的结果写到kafka主题中
        //注意：如果左表数据先到，右表数据后到，kafka主题将会接收到3条消息
        //[左表   null]
        //null
        //[左表   右表]
        //从这样的kakfa主题中读取数据，如果用的是flinkSQL的方式，空消息会自动过滤掉
        //如果使用的是API的方式，默认的SimpleStringSchema是处理不了空消息的，需要自定义反序列化器
        //另外，过滤掉空消息后，剩下的两条数据属于重复数据，在DWS处理额度时候，需要去重
        tableEnv.executeSql("CREATE TABLE emp_dept (\n" +
            "  empno integer,\n" +
            "  ename string,\n" +
            "  deptno integer,\n" +
            "  dname string,\n" +
            "  PRIMARY KEY (empno) NOT ENFORCED\n" +
            ") WITH (\n" +
            "  'connector' = 'upsert-kafka',\n" +
            "  'topic' = 'first',\n" +
            "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
            "  'key.format' = 'json',\n" +
            "  'value.format' = 'json'\n" +
            ")");
        tableEnv.executeSql("insert into emp_dept select e.empno,e.ename,d.deptno,d.dname from emp e left join dept d on e.deptno = d.deptno");

    }
}
